package com.bytecub.simulate.device.mqtt;

import com.bytecub.common.constants.BCConstants;
import com.bytecub.common.domain.dto.response.prop.TemplateResDto;
import com.bytecub.simulate.device.common.constants.SimulateStorage;
import com.bytecub.simulate.device.instance.DeviceInstance;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.DigestUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 *  * ByteCub.cn.
 *  * Copyright (c) 2020-2021 All Rights Reserved.
 *  * 
 *  * @author bytecub@163.com songbin
 *  * @version Id: SimulateMqttClient.java, v 0.1 2021-01-07  Exp $$
 *  
 */
public class SimulateMqttClient {
    private static final Logger logger = LoggerFactory.getLogger(SimulateMqttClient.class);
    private MqttClient mqttClient;
    private String host;
    private String clientId;
    private MqttConnectOptions options = new MqttConnectOptions();
    private Boolean isConnected = false;
    private List<String> topics = new ArrayList<>();
    private String productCode;
    private String deviceCode;
    private String deviceSecret;
    private String upTopic;
    private String eventTopic;
    private String invokeTopic;
    private String invokeReplyTopic;
    private DeviceInstance deviceInstance;

    public SimulateMqttClient(DeviceInstance deviceInstance, String url, String clientId, String productCode,
            String deviceSecret) {
        this.clientId = clientId;
        this.deviceCode = clientId;
        this.productCode = productCode;
        this.deviceInstance = deviceInstance;
        this.host = url;
        this.deviceSecret = deviceSecret;
        this.upTopic = BCConstants.MQTT.GLOBAL_UP_PREFIX + productCode + "/" + deviceCode + "/prop";
        this.eventTopic = BCConstants.MQTT.GLOBAL_UP_PREFIX + productCode + "/" + deviceCode + "/event";
        this.invokeTopic = BCConstants.MQTT.GLOBAL_DOWN_PREFIX + productCode + "/" + deviceCode + "/service/invoke";
        this.invokeReplyTopic = BCConstants.MQTT.GLOBAL_UP_PREFIX + productCode + "/" + deviceCode + "/service/reply";
    }

    public MqttClient getMqttClient() {
        return this.mqttClient;
    }

    public List<String> getTopics() {
        return topics;
    }

    public void setTopics(List<String> topics) {
        this.topics = topics;
    }

    public void genClient() {
        try {
            this.mqttClient = new MqttClient(SimulateStorage.getMqttServerUrl(), this.clientId,
                    new MemoryPersistence());
            this.options.setCleanSession(true);
            this.options.setAutomaticReconnect(true);
            this.options.setConnectionTimeout(SimulateStorage.getTimeout());
            this.options.setKeepAliveInterval(SimulateStorage.getKeepalive());
            String userName = this.deviceCode + "|" + System.currentTimeMillis();
            this.options.setUserName(userName);
            String password = userName + "|" + this.deviceSecret;
            password = DigestUtils.md5DigestAsHex(password.getBytes());
            this.options.setPassword(password.toCharArray());
            try {
                this.mqttClient.connect(options);
                this.mqttClient.setCallback(new SimulateMqttCallback(deviceInstance, this, options, this.host));
                this.isConnected = true;
            } catch (Exception e) {
                logger.warn("连接MQTT服务器", e);
                this.isConnected = false;
                this.tryConnect();
            }
        } catch (Exception e) {
            logger.warn("", e);
            throw new RuntimeException(e);
        }
    }

    public void subscribe(String topic) {
        try {
            this.mqttClient.subscribe(topic);
            this.topics.add(topic);
        } catch (Exception e) {
            logger.warn("订阅主题[{}]异常", topic, e);
            throw new RuntimeException(e);
        }
    }

    public void subscribe(String[] topics) {
        try {
            this.mqttClient.subscribe(topics);
            List<String> array = Arrays.asList(topics);
            this.topics.addAll(array);
        } catch (Exception e) {
            logger.warn("订阅主题[{}]异常", topics, e);
            throw new RuntimeException(e);
        }
    }

    public void subscribe(List<String> topicList) {
        try {
            String[] topics = (String[]) topicList.toArray(new String[topicList.size()]);
            logger.info("订阅主题:{}", topics);
            this.mqttClient.subscribe(topics);
            this.topics.addAll(topicList);
        } catch (Exception e) {
            logger.warn("订阅主题[{}]异常", topics, e);
            throw new RuntimeException(e);
        }
    }

    private void tryConnect() {
        int count = 1;
        while (!this.isConnected) {
            try {
                int sleepTime = (1000 * count) % (1000 * 60 * 10);
                Thread.sleep(sleepTime);
                logger.info("连接[{}]断开，尝试重连第{}次", this.host, count++);
                this.mqttClient.connect(this.options);
                this.isConnected = true;
            } catch (Exception e) {
                logger.warn("首次重连异常");
            }

        }
    }

    /**
     * 发布，默认qos为1，非持久化
     *
     * @param topic
     * @param pushMessage
     */
    public void publish(String topic, byte[] pushMessage) {
        publish(MqttQoS.AT_LEAST_ONCE.value(), false, topic, pushMessage);
    }

    /**
     * 发布
     *
     * @param qos
     * @param retained
     * @param topic
     * @param pushMessage
     */
    public void publish(int qos, boolean retained, String topic, String pushMessage) {
        this.publish(qos, retained, topic, pushMessage.getBytes());
    }

    public void publish(int qos, boolean retained, String topic, byte[] pushMessage) {
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage);
        MqttTopic mTopic = this.mqttClient.getTopic(topic);
        if (null == mTopic) {
            logger.error("topic not exist:{}", topic);
        }
        MqttDeliveryToken token;
        try {
            token = mTopic.publish(message);
            token.waitForCompletion();
        } catch (MqttPersistenceException e) {
            logger.warn("向主题[{}]发布消息异常{}", topic, e);
            throw new RuntimeException("", e);
        } catch (MqttException e) {
            logger.warn("向主题[{}]发布消息异常{}", topic, e);
            throw new RuntimeException(e);
        }
    }
}
